// Copyright 2021-present The Atlas Authors. All rights reserved.
// This source code is licensed under the Apache 2.0 license found
// in the LICENSE file in the root directory of this source tree.

package dm

import (
	"encoding/hex"
	"fmt"
	"strconv"
	"strings"
	"sync"

	"gitee.com/damengde/atlas/sql/internal/sqlx"
	"gitee.com/damengde/atlas/sql/schema"
)

// DefaultDiff provides basic diffing capabilities for MySQL dialects.
// Note, it is recommended to call Open, create a new Driver and use its
// Differ when a database connection is available.
var DefaultDiff schema.Differ = &sqlx.Diff{DiffDriver: &diff{conn: noConn}}

// A diff provides a MySQL implementation for sqlx.DiffDriver.
type diff struct {
	*conn
	// charset to collation mapping.
	// See, internal directory.
	ch2co, co2ch struct {
		sync.Once
		v   map[string]string
		err error
	}
}

// SchemaAttrDiff returns a changeset for migrating schema attributes from one state to the other.
func (d *diff) SchemaAttrDiff(from, to *schema.Schema) []schema.Change {
	return nil
}

// SchemaObjectDiff returns a changeset for migrating schema objects from
// one state to the other.
func (*diff) SchemaObjectDiff(_, _ *schema.Schema) ([]schema.Change, error) {
	return nil, nil
}

// TableAttrDiff returns a changeset for migrating table attributes from one state to the other.
func (d *diff) TableAttrDiff(from, to *schema.Table) ([]schema.Change, error) {
	return nil, nil
}

// ColumnChange returns the schema changes (if any) for migrating one column to the other.
func (d *diff) ColumnChange(fromT *schema.Table, from, to *schema.Column) (schema.ChangeKind, error) {
	return schema.NoChange, nil
}

// IsGeneratedIndexName reports if the index name was generated by the database.
func (d *diff) IsGeneratedIndexName(_ *schema.Table, idx *schema.Index) bool {
	return false
}

// IndexAttrChanged reports if the index attributes were changed.
func (*diff) IndexAttrChanged(from, to []schema.Attr) bool {
	return false
}

// IndexPartAttrChanged reports if the index-part attributes (collation or prefix) were changed.
func (*diff) IndexPartAttrChanged(fromI, toI *schema.Index, i int) bool {
	return false
}

// ReferenceChanged reports if the foreign key referential action was changed.
func (*diff) ReferenceChanged(from, to schema.ReferenceOption) bool {
	return false
}

// Normalize implements the sqlx.Normalizer interface.
func (d *diff) Normalize(from, to *schema.Table) error {
	indexes := make([]*schema.Index, 0, len(from.Indexes))
	for _, idx := range from.Indexes {
		// MySQL requires that foreign key columns be indexed; Therefore, if the child
		// table is defined on non-indexed columns, an index is automatically created
		// to satisfy the constraint.
		// Therefore, if no such key was defined on the desired state, the diff will
		// recommend dropping it on migration. Therefore, we fix it by dropping it from
		// the current state manually.
		if _, ok := to.Index(idx.Name); ok || !keySupportsFK(from, idx) {
			indexes = append(indexes, idx)
		}
	}
	from.Indexes = indexes

	// In case the "current" state was inspected (or loaded) with the collation/charset attributes,
	// but there are not found on the desired state, detect what are the default settings for the
	// desired state of the table (based on database default) to avoid proposing unnecessary changes.
	if sqlx.Has(from.Attrs, &schema.Collation{}) {
		if err := d.defaultCollate(&to.Attrs); err != nil {
			return err
		}
	}
	if sqlx.Has(from.Attrs, &schema.Charset{}) {
		if err := d.defaultCharset(&to.Attrs); err != nil {
			return err
		}
	}
	return nil
}

// FindTable implements the DiffDriver.TableFinder method in order to provide
// tables lookup that respect the "lower_case_table_names" system variable.
func (d *diff) FindTable(s *schema.Schema, name string) (*schema.Table, error) {
	switch d.lcnames {
	// In mode 0: tables are stored as specified, and comparisons are case-sensitive.
	case 0:
		t, ok := s.Table(name)
		if !ok {
			return nil, &schema.NotExistError{Err: fmt.Errorf("table %q was not found", name)}
		}
		return t, nil
	// In mode 1: the table are stored in lowercase, but they are still
	// returned on inspection, because comparisons are not case-sensitive.
	// In mode 2: the tables are stored as given but compared in lowercase.
	// This option is not supported by Linux-based systems.
	case 1, 2:
		var matches []*schema.Table
		for _, t := range s.Tables {
			if strings.ToLower(name) == strings.ToLower(t.Name) {
				matches = append(matches, t)
			}
		}
		switch n := len(matches); n {
		case 0:
			return nil, &schema.NotExistError{Err: fmt.Errorf("table %q was not found", name)}
		case 1:
			return matches[0], nil
		default:
			return nil, fmt.Errorf("%d matches found for table %q", n, name)
		}
	default:
		return nil, fmt.Errorf("unsupported 'lower_case_table_names' mode: %d", d.lcnames)
	}
}

// collationChange returns the schema change for migrating the collation if
// it was changed, and it is not the default attribute inherited from its parent.
func (*diff) collationChange(from, top, to []schema.Attr) schema.Change {

	return noChange
}

// engineChange returns the schema change for migrating the table engine in case
// it was changed.
func (*diff) engineChange(from, to []schema.Attr) schema.Change {

	return noChange
}

// charsetChange returns the schema change for migrating the collation if
// it was changed, and it is not the default attribute inherited from its parent.
func (*diff) charsetChange(from, top, to []schema.Attr) schema.Change {

	return noChange
}

// columnCharsetChange indicates if there is a change to the column charset.
func (d *diff) columnCharsetChanged(fromT *schema.Table, from, to *schema.Column) (bool, error) {
	return false, nil
}

// columnCollateChanged indicates if there is a change to the column charset.
func (d *diff) columnCollateChanged(fromT *schema.Table, from, to *schema.Column) (bool, error) {
	return false, nil

}

// autoIncChange returns the schema change for changing the AUTO_INCREMENT
// attribute in case it is not the default.
func (*diff) autoIncChange(from, to []schema.Attr) schema.Change {

	return noChange
}

// indexType returns the index type from its attribute.
// The default type is BTREE if no type was specified.
func indexType(attr []schema.Attr) *IndexType {
	t := &IndexType{T: IndexTypeBTree}
	if sqlx.Has(attr, t) {
		t.T = strings.ToUpper(t.T)
	}
	return t
}

// enforced returns the ENFORCED attribute for the CHECK
// constraint. A CHECK is ENFORCED if not state otherwise.
func enforced(attr []schema.Attr) bool {
	if e := (Enforced{}); sqlx.Has(attr, &e) {
		return e.V
	}
	return true
}

// noChange describes a zero change.
var noChange struct{ schema.Change }

func (d *diff) typeChanged(from, to *schema.Column) (bool, error) {
	return false, nil
}

// defaultChanged reports if the default value of a column was changed.
func (d *diff) defaultChanged(from, to *schema.Column) (bool, error) {
	return false, nil
}

// generatedChanged reports if the generated expression of a column was changed.
func (*diff) generatedChanged(from, to *schema.Column) (bool, error) {
	return false, nil
}

// equalIntValues report if the 2 int default values are ~equal.
// Note that default expression are not supported atm.
func (d *diff) equalIntValues(x1, x2 string) bool {
	x1 = strings.ToLower(strings.Trim(x1, "' "))
	x2 = strings.ToLower(strings.Trim(x2, "' "))
	if x1 == x2 {
		return true
	}
	d1, err := strconv.ParseInt(x1, 10, 64)
	if err != nil {
		// Numbers are rounded down to their nearest integer.
		f, err := strconv.ParseFloat(x1, 64)
		if err != nil {
			return false
		}
		d1 = int64(f)
	}
	d2, err := strconv.ParseInt(x2, 10, 64)
	if err != nil {
		// Numbers are rounded down to their nearest integer.
		f, err := strconv.ParseFloat(x2, 64)
		if err != nil {
			return false
		}
		d2 = int64(f)
	}
	return d1 == d2
}

// equalFloatValues report if the 2 float default values are ~equal.
// Note that default expression are not supported atm.
func (d *diff) equalFloatValues(x1, x2 string) bool {
	x1 = strings.ToLower(strings.Trim(x1, "' "))
	x2 = strings.ToLower(strings.Trim(x2, "' "))
	if x1 == x2 {
		return true
	}
	d1, err := strconv.ParseFloat(x1, 64)
	if err != nil {
		return false
	}
	d2, err := strconv.ParseFloat(x2, 64)
	if err != nil {
		return false
	}
	return d1 == d2
}

// equalsStringValues report if the 2 string default values are
// equal after dropping their quotes.
func equalsStringValues(x1, x2 string) bool {
	a, err1 := sqlx.Unquote(x1)
	b, err2 := sqlx.Unquote(x2)
	return a == b && err1 == nil && err2 == nil
}

// boolValue returns the MySQL boolean value from the given string (if it is known).
func boolValue(x string) (bool, error) {
	switch x {
	case "1", "'1'", "TRUE", "true":
		return true, nil
	case "0", "'0'", "FALSE", "false":
		return false, nil
	default:
		return false, fmt.Errorf("mysql: unknown value: %q", x)
	}
}

// binValue returns the MySQL binary value from the given string (if it is known).
func binValue(x string) (string, error) {
	if !isHex(x) {
		return x, nil
	}
	d, err := hex.DecodeString(x[2:])
	if err != nil {
		return x, err
	}
	return string(d), nil
}

// keySupportsFK reports if the index key was created automatically by MySQL
// to support the constraint. See sql/sql_table.cc#find_fk_supporting_key.
func keySupportsFK(t *schema.Table, idx *schema.Index) bool {
	return false
}

// defaultCollate appends the default COLLATE to the attributes in case a
// custom character-set was defined for the element and the COLLATE was not.
func (d *diff) defaultCollate(attrs *[]schema.Attr) error {
	return nil
}

// defaultCharset appends the default CHARSET to the attributes in case a
// custom collation was defined for the element and the CHARSET was not.
func (d *diff) defaultCharset(attrs *[]schema.Attr) error {
	return nil
}
